package com.zyh.flink.day03;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Map;

public class MapStateTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> hadoop10 = environment.socketTextStream("hadoop10", 9999);
        KeyedStream<Tuple2<String, String>, String> keyedStream = hadoop10.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String s) throws Exception {
                String[] ss = s.split("\\s+");
                return Tuple2.of(ss[0], ss[1]);
            }
        }).keyBy(t -> t.f0);
        SingleOutputStreamOperator<String> result = keyedStream.map(new MyMapStateFunction());
        result.print();

        environment.execute("MapStateJob");
    }
}
//通过MapState记录用户访问的类别和次数
class MyMapStateFunction extends RichMapFunction<Tuple2<String,String>,String>{
    private MapState<String,Integer> mapState;

    @Override
    public void open(Configuration parameters) throws Exception {
        RuntimeContext ctx = getRuntimeContext();
        MapStateDescriptor<String, Integer> msd = new MapStateDescriptor<>("categories", Types.STRING, Types.INT);
        this.mapState = ctx.getMapState(msd);
    }

    /*
    * value 表示一个用户访问的一次类别
    * */
    @Override
    public String map(Tuple2<String, String> value) throws Exception {
        //使用mapState记录访问的类别历史
        //首先判断map中是否已经存在相同的key
        if (this.mapState.contains(value.f1)){
            //存在则更新map中的value部分+1
            this.mapState.put(value.f1, this.mapState.get(value.f1)+1);
        }else {
            //不存在则在map中添加(key,1)
            this.mapState.put(value.f1, 1);
        }
        //遍历mapState中所有类别拼接成字符串返回
        StringBuilder stringBuilder = new StringBuilder(value.f0);
        for (Map.Entry<String, Integer> entry : this.mapState.entries()) {
            String k = entry.getKey();
            Integer v = entry.getValue();
            stringBuilder.append(" ").append(k).append("-").append(v);
        }

        return stringBuilder.toString();
    }
}